// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/functions/remote/utils/remote_function_service_provider.h>

#include "thrift/lib/cpp2/server/ThriftServer.h"
#include <pollux/functions/remote/server/remote_function_service.h>

namespace kumo::pollux::functions {
    RemoteFunctionServiceParams
    RemoteFunctionServiceProviderForLocalThrift::getRemoteFunctionServiceParams() {
        melon::call_once(initializeServiceFlag_, [&]() { initializeServer(); });
        return RemoteFunctionServiceParams{
            remotePrefix_,
            location_,
        };
    }

    void RemoteFunctionServiceProviderForLocalThrift::initializeServer() {
        auto handler =
                std::make_shared<pollux::functions::RemoteFunctionServiceHandler>(
                    remotePrefix_);
        server_ = std::make_shared<apache::thrift::ThriftServer>();
        server_->setInterface(handler);
        server_->setAddress(location_);

        thread_ = std::make_unique<std::thread>([&] { server_->serve(); });
        POLLUX_CHECK(waitForRunning(), "Unable to initialize thrift server.");
        KLOG(INFO) << "Thrift server is up and running in local port " << location_;
    }

    bool RemoteFunctionServiceProviderForLocalThrift::waitForRunning() {
        for (size_t i = 0; i < 100; ++i) {
            if (server_->getServerStatus() ==
                apache::thrift::ThriftServer::ServerStatus::RUNNING) {
                return true;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }
        return false;
    }

    RemoteFunctionServiceProviderForLocalThrift::
    ~RemoteFunctionServiceProviderForLocalThrift() {
        server_->stop();
        thread_->join();
        KLOG(INFO) << "Thrift server stopped.";
    }

    RemoteFunctionServiceParams startLocalThriftServiceAndGetParams() {
        static melon::Singleton<IRemoteFunctionServiceProvider>
                remoteFunctionServiceProviderForLocalThriftSingleton{
                    []() { return new RemoteFunctionServiceProviderForLocalThrift(); }
                };
        auto provider =
                remoteFunctionServiceProviderForLocalThriftSingleton.try_get();
        POLLUX_CHECK(provider, "local remoteFunctionProvider is not available");
        return provider->getRemoteFunctionServiceParams();
    }
} // namespace kumo::pollux::functions
